import time,re
import pandas as pd
import redis
import pickle
import logging
import pywencai
import akshare as ak
import pandas as pd
import configparser
import datetime
from sqlalchemy import create_engine
import pymysql
import os
import traceback
from 个股行业与概念涨幅分析回测 import get_brand_and_concept_rank



pymysql.install_as_MySQLdb()


log_format = "%(asctime)s - %(levelname)s - %(process)d - %(filename)s:%(lineno)d - %(message)s"
date_format = "%Y-%m-%d %H:%M:%S"  # 精确到秒
logging.basicConfig(level=logging.DEBUG, format=log_format, datefmt=date_format)

pid = os.getpid()
# query_date = datetime.datetime.now().strftime('%Y%m%d')

# 日志文件路径
log_file_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), f'log/{pid}.log')

# 创建一个 handler，用于写入日志文件
file_handler = logging.FileHandler(log_file_path)
file_handler.setFormatter(logging.Formatter(log_format, date_format))
# 添加 handler 到 logger
logging.getLogger().addHandler(file_handler)

# 初始化配置解析器
config = configparser.ConfigParser()

# 读取配置文件
import os
current_dir = os.path.dirname(os.path.abspath(__file__))
config.read(current_dir+'/config.ini', encoding='utf-8')


# 获取Redis的配置信息
redis_host = config.get('Redis', 'host')
redis_port = config.getint('Redis', 'port')
redis_db = config.getint('Redis', 'db')
redis_password = config.get('Redis', 'password')
r = redis.Redis(host=redis_host, port=redis_port, db=redis_db, password=redis_password)

mysql_port = config.getint('mysql', 'port')
mysql_host = config.get('mysql', 'host')
mysql_db = config.get('mysql', 'db')
import urllib.parse
mysql_password = urllib.parse.quote(config.get('mysql', 'password'))
mysql_user = config.get('mysql', 'user')
db_url = f'mysql://{mysql_user}:{mysql_password}@{mysql_host}:{mysql_port}/{mysql_db}'

engine = create_engine(db_url,pool_size=20,max_overflow=20,pool_recycle=60)


# subscriber = r.pubsub()
# subscriber.subscribe('bjzt_channel')

# query_date = datetime.datetime.now().strftime('%Y-%m-%d')

def time_decorator(func):
    def wrapper(*args, **kwargs):
        start_time = time.time()  # 记录函数开始执行的时间
        result = func(*args, **kwargs)  # 调用函数
        end_time = time.time()  # 记录函数执行完毕的时间
        print(f"{func.__name__} executed in {end_time - start_time:.4f} seconds")
        return result
    return wrapper
def get_next_trade_date(trade_date):
    tool_trade_date_hist_sina_df = ak.tool_trade_date_hist_sina()
    # 筛选出所有晚于给定交易日的日期
    date_df = tool_trade_date_hist_sina_df[tool_trade_date_hist_sina_df["trade_date"] > pd.Timestamp(trade_date).date()]
    date_df = date_df.sort_values(by="trade_date", ascending=True)  # 按日期升序排序
    next_trade_date = date_df["trade_date"].values[0]  # 获取最接近给定日期的下一个交易日
    return next_trade_date.strftime("%Y%m%d")  # 格式化日期

# 定义获取前一个交易日的函数
def get_pre_trade_date(trade_date):
    tool_trade_date_hist_sina_df = ak.tool_trade_date_hist_sina()
    date_df = tool_trade_date_hist_sina_df[tool_trade_date_hist_sina_df["trade_date"] < pd.Timestamp(trade_date).date()]
    date_df = date_df.sort_values(by="trade_date", ascending=False)
    pre_trade_date = date_df["trade_date"].values[0]
    return pre_trade_date.strftime("%Y%m%d")

def get_pre_trade_date_n(trade_date,n):
    tool_trade_date_hist_sina_df = ak.tool_trade_date_hist_sina()
    date_df = tool_trade_date_hist_sina_df[tool_trade_date_hist_sina_df["trade_date"] < pd.Timestamp(trade_date).date()]
    date_df = date_df.sort_values(by="trade_date", ascending=False)
    pre_trade_date = date_df["trade_date"].values[n-1]
    return pre_trade_date.strftime("%Y%m%d")



def analysis(stock_name,test_info):
    try:
        logging.info(f"开始监控：{stock_name}")
        stock_fupan_df = test_info["stock_fupan_df"]



        real_stock_info_tdx_list = test_info["real_market_info_tdx"]

        # logging.info(real_stock_info_tdx_list)

        real_market_concept_index_tdx_list = test_info["real_market_concept_index_tdx"]
        real_market_brand_index_tdx_list = test_info["real_market_brand_index_tdx"]
        stock_rps_df = test_info["stock_rps_df"]

        stock_fupan_df = stock_fupan_df[stock_fupan_df['股票简称'] == stock_name]
        market_value = stock_fupan_df['流通市值'].values[0]
        logging.info(f"市值:{market_value}亿")


        lianban = stock_fupan_df['连续涨停天数'].values[0]
        logging.info(f"连续涨停天数:{lianban}")
        code = stock_fupan_df["股票代码"].values[0]
    except Exception as e:
        tb_info = traceback.format_exc()
        # 将异常信息和 traceback 信息一起记录
        logging.info(f"An error occurred: {e}\nTraceback info:\n{tb_info}")
        return False,"获取股票信息失败"

    is20 = False


    if "ST" in stock_name:
        return False,"ST股票，退出"

    if lianban >= 1:
        return False,"连板股票，退出"


    if code.startswith("30") or code.startswith("688"):
        is20 = True
        # return False,"20cm标的，退出"

    # if r.get(f"buy_stock:{stock_name}") == "买入":
    #     return False,"已经买入，退出"
    time_index = 0

    while True:
        r.set(f"stock_analysis_test:{stock_name}", "分析中")
        r.expire(f"stock_analysis_test:{stock_name}", 3)

        real_stock_info_tdx = pd.DataFrame(real_stock_info_tdx_list[time_index])
        real_market_brand_index_tdx = pd.DataFrame(real_market_brand_index_tdx_list[time_index])
        real_market_concept_index_tdx = pd.DataFrame(real_market_concept_index_tdx_list[time_index])


        start_time = time.time()
        try:
            time_index = time_index + 1



            real_stock_info_tdx["amount"] = real_stock_info_tdx["amount"]/10000
            df = pd.merge(stock_fupan_df,real_stock_info_tdx,how = "inner",left_on=["股票代码"], right_on=["code"])
            df = df[df["股票简称"] == stock_name]

            df = pd.merge(df, stock_rps_df, how="inner", left_on=["股票简称"], right_on=["股票简称"])



            stock_info={}
            stock_info["code"] = df["code"].values[0]
            stock_info["name"]= df["股票简称"].values[0]
            stock_info["amount"] = round(df["amount"].values[0],2)
            stock_info["change"] = round(df["change"].values[0],2)
            stock_info["price"] = round(df["price"].values[0], 2)
            stock_info["vol"] =  round(df["vol"].values[0], 2)
            stock_info["yesterday_amount"] = round(df["成交额"].values[0], 2)
            stock_info["market_value"] = round(df["流通市值"].values[0], 2)
            stock_info["hsl"] = round(stock_info["amount"]*100/stock_info["market_value"],2)
            stock_info["zhsl"] = round(stock_info["amount"]*100/stock_info["yesterday_amount"],2)
            stock_info["open"] = round(df["open"].values[0], 2)
            stock_info["last_close"] = round(df["last_close"].values[0], 2)
            stock_info["avr_change"] = round(df["avr_change"].values[0], 2)
            stock_info["bias5"] = round(df["5日乖离"].values[0], 2)
            stock_info["bias10"] = round(df["10日乖离"].values[0], 2)
            stock_info["bias20"] = round(df["20日乖离"].values[0], 2)
            stock_info["zt_1_year"] = df["近一年涨停次数"].values[0]
            stock_info["rps20"] = df["rps20"].values[0]
            stock_info["timestamp"] = df["timestamp"].values[0]




            stock_info["pid"] = pid



            #买入时的判断参数
            if is20==False:
                zhsl_low = 60
                zhsl_high = 125
                change_low = 6
                change_high = 9
                avr_change_low = 5
                avr_change_diff = 3
                bais10 = 99
                bais5 = 99
                market_value_min = 15
                market_value_max = 200
                amount = 0.3
                concept_rps = 0
                suggestion="板块10cm领涨，买"
            else:
                bais10 = 10
                bais5 = 10
                zhsl_low = 60*1.5
                zhsl_high = 125*1.5
                change_low = 15
                change_high = 18
                avr_change_low = 10
                avr_change_diff = 3
                market_value_min = 15
                market_value_max =200
                amount = 0.3
                concept_rps = 0
                suggestion = "板块20cm领涨，买"


            brand_and_concept_dict = get_brand_and_concept_rank(test_info["query_time"],stock_info["name"],stock_fupan_df,real_stock_info_tdx,real_market_brand_index_tdx,real_market_concept_index_tdx)
            if brand_and_concept_dict is None:
                return False, "没有概念板块信息，不进行分析"
            stock_info = {**stock_info, **brand_and_concept_dict}


            # zhls_diff = 10
            # if datetime.datetime.strptime(stock_info["timestamp"], '%Y-%m-%d %H:%M:%S') < datetime.datetime.now().replace(hour=10, minute=0, second=0, microsecond=0):
            #     zhsl_high = 90+zhls_diff
            # elif datetime.datetime.strptime(stock_info["timestamp"], '%Y-%m-%d %H:%M:%S') < datetime.datetime.now().replace(hour=11, minute=0, second=0, microsecond=0):
            #     zhsl_high = 100+zhls_diff
            # elif datetime.datetime.strptime(stock_info["timestamp"], '%Y-%m-%d %H:%M:%S') < datetime.datetime.now().replace(hour=14, minute=0, second=0, microsecond=0):
            #     zhsl_high = 110+zhls_diff


            #排除已经涨停的票
            # if stock_info["change"] >9.8 and is20==False:
            #     return False,"已经10cm涨停"
            #
            # if stock_info["change"] >19.8 and is20==True:
            #     return False,"已经20cm涨停"

            #200秒内没有异动，退出
            # if time_index > 200:
            #     return False,"时间超过200秒，退出"


            """
            买入约束条件：
            1、主板在7-9个点之间，创业板在13-19点之间
            2、自由换手不能太大，太大表明卖的多，是个弱板，涨停时接近昨日成交额较好或者缩量较好。
            3、概念板块所在的涨幅排名不能太低，换算成百分比，在95左右比较好。概念取所有概念中，涨幅第一的概念，作为当日该股的主要炒作概念。
            4、月乖离不能太大，太大容易发生冲高回落。stock_info["bias20"] > 100这个主要是针对没有月线的新股设置的
            5、同一概念中，只做前排个股。对于主板个股，剔除一字板和20cm的个股后，如果涨幅排名还是第一名，则买入。创业板个股，只买涨幅排名第一的个股，其它情况全部放弃。
            6、暂定买入市值大于10亿的股票
            7、暂定买入已成交金额4000万以上的股票
            """


            if (stock_info["change"] > change_low
                    and stock_info["change"] < change_high
                    and stock_info["zhsl"] > zhsl_low
                    and stock_info["zhsl"] < zhsl_high
                    and stock_info["amount"]>amount
                    and (stock_info["bias5"] < bais5 or stock_info["bias5"] > 100)
                    # and (stock_info["bias10"] < bais10 or stock_info["bias10"] > 100)
                    and stock_info["concept_rps_hs"]>=concept_rps
                    and stock_info["market_value"]>market_value_min
                    and stock_info["market_value"] < market_value_max
                    and stock_info["suggestion"]==suggestion
                    and stock_info["rps20"]<99
                    and time_index <180):



                key = f"stock_buy_{test_info['query_time']}:{stock_info['name']}"
                if r.get(key) is None:
                    stock_info["desc"] = "买入"
                    pd.DataFrame([stock_info]).to_sql("analysis_stock_step_test", engine, if_exists='append', index=False)
                    s = f"{stock_info['name']}满足条件，买入"

                    return True, s

                else:
                    s = f"{stock_info['name']}已经买入"
                    logging.info(s)
                    return False, s

            else:
                unsatisfied_conditions = []
                if not (stock_info["change"] > change_low):
                    unsatisfied_conditions.append("涨幅小于最小值")
                if not (stock_info["change"] < change_high):
                    unsatisfied_conditions.append("涨幅大于最大值")
                if not (stock_info["zhsl"] > zhsl_low):
                    unsatisfied_conditions.append("自由换手太小")
                if not (stock_info["zhsl"] < zhsl_high):
                    unsatisfied_conditions.append("自由换手太大")
                if not (stock_info["amount"] > amount):
                    unsatisfied_conditions.append("交易额太小")
                if not (stock_info["bias5"] < bais5 or stock_info["bias5"] > 100):
                    unsatisfied_conditions.append("5日乖离率太大")
                # if not (stock_info["bias10"] < bais10 or stock_info["bias10"] > 100):
                #    unsatisfied_conditions.append("bias10 < bais10 or bias10 > 100")
                if not (stock_info["concept_rps_hs"] > concept_rps):
                    unsatisfied_conditions.append("板块强度太小")
                if not (stock_info["market_value"] > market_value_min):
                    unsatisfied_conditions.append("市值太小")
                if not (stock_info["market_value"] < market_value_max):
                    unsatisfied_conditions.append("市值太大")
                if not (stock_info["suggestion"] != '不买'):
                    unsatisfied_conditions.append("板块内非前排")
                if not (stock_info["rps20"] < 99):
                    unsatisfied_conditions.append("个股rps值太大")
                if not (time_index < 200):
                    unsatisfied_conditions.append("监控时间超过200秒")

                stock_info["desc"] = "; ".join(unsatisfied_conditions)


            logging.info(stock_info)
            pd.DataFrame([stock_info]).to_sql("analysis_stock_step_test", engine, if_exists='append', index=False)

            if time_index > 80:
                return False,"时间超过80秒，退出"

            if stock_info["change"] >9.8 and is20==False:
                return False,"已经10cm涨停"

            if stock_info["change"] >19.8 and is20==True:
                return False,"已经20cm涨停"

            if stock_info["rps20"] > 99:
                return False,"个股rps值太大"

        except Exception as e:
            tb_info = traceback.format_exc()
            s = f"An error occurred: {e}\nTraceback info:\n{tb_info}"
            logging.info(s)
            return False, s


def get_time_200_seconds_later(start_time_str):
    from datetime import datetime, timedelta
    start_time = datetime.strptime(start_time_str, "%Y-%m-%d %H:%M:%S")
    # 计算开始时间200秒后的时刻
    time_200_seconds_later = start_time + timedelta(seconds=200)
    return time_200_seconds_later.strftime("%Y-%m-%d %H:%M:%S")
@time_decorator
def get_test_info(name,query_time,query_time_20):
    t = {}

    end_time = get_time_200_seconds_later(query_time)
    datetime_obj = datetime.datetime.strptime(query_time, "%Y-%m-%d %H:%M:%S")
    formatted_date = datetime_obj.strftime("%Y%m%d")
    t["query_time"] = formatted_date
    #获取盘前数据
    # pre_20_day = get_pre_trade_date_n(formatted_date,20)

    logging.info("获取RPS数据")
    sql = f'''
        SELECT 股票简称, MAX(`rps20`) AS rps20
        FROM stock_rps
        WHERE trade_date >= '{query_time_20}' and trade_date<='{formatted_date}'
        GROUP BY 股票简称;
        '''
    stock_rps_df = pd.read_sql(sql, engine)
    t["stock_rps_df"] = stock_rps_df


    #获取盘前数据
    logging.info("获取盘前数据")
    sql = f'''
            select * from stock_fupan sf where 日期='{formatted_date}' and 股票简称='{name}' 
            '''
    stock_fupan_df = pd.read_sql(sql, engine)
    t["stock_fupan_df"] = stock_fupan_df

    code = stock_fupan_df["股票代码"].values[0]

    #获取交易数据
    logging.info("获取交易数据real_market_info_tdx")
    sql = f'''
        select * from stock.real_market_info_tdx_h where `timestamp` >='{query_time}' and `timestamp`<='{end_time}' 
        '''
    logging.info(sql)
    real_market_info_tdx_df = pd.read_sql(sql, engine)
    real_market_info_tdx_df.sort_values(by='timestamp', inplace=True)
    grouped = real_market_info_tdx_df.groupby('timestamp')
    t["real_market_info_tdx"] = [group.copy() for _, group in grouped]


    #获取交易数据
    logging.info("real_market_concept_index_tdx")
    sql = f'''
        select * from stock.real_market_concept_index_tdx_h where `timestamp` >='{query_time}' and `timestamp`<='{end_time}' 
        '''

    real_market_concept_index_tdx_h_df = pd.read_sql(sql, engine)
    real_market_concept_index_tdx_h_df.sort_values(by='timestamp', inplace=True)
    grouped = real_market_concept_index_tdx_h_df.groupby('timestamp')
    t["real_market_concept_index_tdx"] = [group.copy() for _, group in grouped]


    #获取交易数据
    logging.info("eal_market_brand_index_tdx")
    sql = f'''
        select * from stock.real_market_brand_index_tdx_h where `timestamp` >='{query_time}' and `timestamp`<='{end_time}'
        '''

    real_market_brand_index_tdx_h_df = pd.read_sql(sql, engine)
    real_market_brand_index_tdx_h_df.sort_values(by='timestamp', inplace=True)
    grouped = real_market_brand_index_tdx_h_df.groupby('timestamp')
    t["real_market_brand_index_tdx"] = [group.copy() for _, group in grouped]


    return t

def get_dxjl_stock(query_date):
    sql = f'''
            select * from real_market_info_dxjl_h rmidh where trade_type in ('笼子触涨停','逼近涨停') and Date(`timestamp`) ='{query_date}'
        '''
    print(sql)
    df_dxjl = pd.read_sql(sql, engine)
    df_dxjl = df_dxjl.sort_values(by="timestamp",ascending=True)
    return df_dxjl




if __name__ == "__main__":
    logging.info("监控进程启动")

    while True:
        try:
            queue, message = r.brpop("bjzt_channel_test")

            row = pickle.loads(message)
            print(row)


            timestamp_str = row['timestamp'].strftime('%Y-%m-%d %H:%M:%S')
            test_info = get_test_info(row["stock_name"],timestamp_str,row["query_time_20"])


            if r.get(f"stock_analysis_test:{row["stock_name"]}") is not None:
                logging.info("已经在监控")
                continue

            res, desc = analysis(row["stock_name"],test_info)
            if res == False:
                logging.info(f"row['stock_name']:{desc}")
            logging.info(f"row['stock_name']:{desc}")

        except Exception as e:
            tb_info = traceback.format_exc()
            logging.info(f"An error occurred: {e}\nTraceback info:\n{tb_info}")


    # query_time = "2024-04-26"
    #
    # df_dxjl = get_dxjl_stock(query_time)
    # print(df_dxjl)
    #
    # for index, row in df_dxjl.iterrows():
    #     timestamp_str = row['timestamp'].strftime('%Y-%m-%d %H:%M:%S')
    #     test_info = get_test_info(row["stock_name"],timestamp_str)
    #     try:
    #         res, desc = analysis(row["stock_name"],test_info)
    #         if res == False:
    #             logging.info(f"row['stock_name']:{desc}")
    #         logging.info(f"row['stock_name']:{desc}")
    #
    #     except Exception as e:
    #         tb_info = traceback.format_exc()
    #         logging.info(f"An error occurred: {e}\nTraceback info:\n{tb_info}")







